package nuist.hanniko.stu.webbackend.service.cmd_runner;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import nuist.hanniko.stu.webbackend.model.LwM2m.request.*;
import nuist.hanniko.stu.webbackend.service.impls.DeviceService;
import nuist.hanniko.stu.webbackend.util.RandomNameGen;
import nuist.hanniko.stu.webbackend.util.RequestSerializer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.CommandLineRunner;
import org.springframework.core.annotation.Order;
import org.springframework.data.redis.connection.MessageListener;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.listener.ChannelTopic;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;
import org.springframework.stereotype.Component;

import java.util.HashSet;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.*;

@Component
@Order(value = 1)
public class DeviceRequestService implements CommandLineRunner {

    private final RedisMessageListenerContainer container;
    private final StringRedisTemplate redisTemplate;

    private static final Logger LOG = LoggerFactory.getLogger(DeviceService.class);
    private static final String REQ_CHANNEL = "LESHAN_REQ";
    private static final String RESP_CHANNEL = "LESHAN_RESP";
    private static final long TIMEOUT = 15;     // timeout, unit - second

    private final Set<String> tickets = new HashSet<>();
    private final ExecutorService executorService = Executors.newCachedThreadPool();
    private final Map<String, BlockingQueue<JSONObject>> ticketToBlockingQueue = new ConcurrentHashMap<>();

    // resp 监听器
    private final MessageListener responseListener = (message, bytes) -> {
        JSONObject jsonObject = (JSONObject) JSON.parse(message.toString());
        String ticket = jsonObject.getString("ticket");

        LOG.info("[{}]: {}", ticket, jsonObject.toString());

        if (ticketToBlockingQueue.containsKey(ticket)) {
            ticketToBlockingQueue.get(ticket).offer(jsonObject);
        }
    };

    @Autowired
    public DeviceRequestService(RedisMessageListenerContainer container, StringRedisTemplate redisTemplate) {
        this.container = container;
        this.redisTemplate = redisTemplate;
    }

    @Override
    public void run(String... args) throws Exception {
        LOG.info("-------- 请求监听器 - 启动 --------");

        container.addMessageListener(responseListener, new ChannelTopic(RESP_CHANNEL));
    }

    /* 1. generate ticket attached to request
       2. send request to REQ_CHANNEL
       3. get response from RESP_CHANNEL
       4. return response
     */
    public JSONObject sendRequest(BaseRequest request) {

        JSONObject response = null;

        // serialize request to json object
        JSONObject requestObject = RequestSerializer.serialize2Json(request);

        // generate ticket
        String ticket = RandomNameGen.getRandomString();
        while (tickets.contains(ticket)) {
            ticket = RandomNameGen.getRandomString();
        }
        tickets.add(ticket);
        requestObject.put("ticket", ticket);

        // create internal blocking queue according to ticket
        ticketToBlockingQueue.put(ticket, new ArrayBlockingQueue<>(2));

        try {
            // publish data to request channel
            publish(requestObject);

            // get response from blocking queue
            JSONObject tmpResp = ticketToBlockingQueue.get(ticket).poll(TIMEOUT, TimeUnit.SECONDS);
            LOG.info(tmpResp.toString());

            if (tmpResp.containsKey("ack")) {
                // continue to listen for response body
                response = ticketToBlockingQueue.get(ticket).poll(TIMEOUT, TimeUnit.SECONDS);
            } else if (tmpResp.containsKey("err")) {
                response = tmpResp;
            }

        } catch (InterruptedException e) {
            LOG.error(e.getMessage());
        } finally {
            // remove ticket from blockingQueue map and hashSet
            ticketToBlockingQueue.remove(ticket);
            tickets.remove(ticket);
        }

        return response;
    }

    private void publish(JSONObject requestObject) {
        this.executorService.submit(() ->
                redisTemplate.convertAndSend(REQ_CHANNEL, requestObject.toString()));
    }


}
